package com.warm.pump.executor.batch;

import com.alibaba.fastjson.JSON;
import com.warm.pump.executor.InstanceService;
import com.warm.pump.frame.db.DbUtil;
import com.warm.pump.frame.gen.util.TypeCast;
import com.warm.pump.frame.util.StringUtils;
import com.warm.pump.module.skd.bean.po.gen.ScheduleInstanceDtl;
import com.warm.pump.module.skd.dao.gen.ScheduleInstanceDtlDao;
import com.warm.pump.module.work.bean.po.gen.Work;
import com.warm.pump.module.work.bean.po.gen.WorkItem;
import com.warm.pump.module.work.bean.po.gen.WorkItemSub;
import com.warm.pump.module.work.dao.gen.WorkDao;
import com.warm.pump.module.work.dao.gen.WorkItemDao;
import com.warm.pump.module.work.dao.gen.WorkItemSubDao;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.*;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.List;
import java.util.Map;


/**
 * Created by guozj on 2018/8/22.
 */
@Configuration
@EnableBatchProcessing
public class BatchConfig {
    private static final Logger logger = LoggerFactory.getLogger(BatchConfig.class);
    @Autowired
    private WorkItemDao workItemDao;
    @Autowired
    private WorkDao workDao;
    @Autowired
    private WorkItemSubDao workItemSubDao;
    @Autowired
    private ScheduleInstanceDtlDao scheduleInstanceDtlDao;
    @Autowired
    private InstanceService instanceService;

    /**
     * 自定义reader  添加注解StepScope 让其生命周期只为 一个batch step
     *
     * @param itemId
     * @param params
     * @return CursorItemReader 返回类型 不能为ItemReader
     * @throws Exception
     */
    @Bean
    @StepScope
    public CursorItemReader cursorItemReader(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['params']}") String params) throws Exception {
        CursorItemReader itemReader = new CursorItemReader();


        if (null == itemId) {
            return null;
        }
        WorkItem workItemDtl = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));
        Map<String, Object> condValueMap;
        if (StringUtils.isEmpty(params) || "null".equals(params)) {
            condValueMap = instanceService.getParamsMap(workItemDtl.getWorkId());
        } else {
            condValueMap = JSON.parseObject(params);
        }
        itemReader.setCondValueMap(condValueMap);
        itemReader.setItemDtl(workItemDtl);
        itemReader.setDataSource(DbUtil.getDataSource(workItemDtl.getDbKeyFrom()));
        return itemReader;
    }

    /**
     * 自定义Processor
     *
     * @param itemId
     * @param params
     * @return
     */
    @Bean
    @StepScope
    public CursorProcessor cursorProcessor(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['params']}") String params) {
        CursorProcessor processor = new CursorProcessor();
        if (null == itemId) {
            return null;
        }
        WorkItem workItemDtl = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));

        Map<String, Object> subCondMap = new HashMap<>();
        subCondMap.put("itemDtlId", workItemDtl.getId());
        subCondMap.put("orderBy", "order_num");
        List<WorkItemSub> subList = workItemSubDao.getListByMap(subCondMap);

        Map<String, Object> condValueMap;
        if (StringUtils.isEmpty(params) || "null".equals(params)) {
            condValueMap = instanceService.getParamsMap(workItemDtl.getWorkId());
        } else {
            condValueMap = JSON.parseObject(params);
        }
        processor.setCondValueMap(condValueMap);
        processor.setItemDtl(workItemDtl);
        processor.setSubList(subList);
        return processor;
    }

    /**
     * 自定义write
     *
     * @param itemId
     * @param params
     * @return
     */
    @Bean
    @StepScope
    public CursorWriter cursorWriter(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['params']}") String params) {
        CursorWriter writer = new CursorWriter();
        if (null == itemId) {
            return null;
        }
        WorkItem workItem = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));
        writer.setItemDtl(workItem);
        return writer;
    }

    @Bean
    @StepScope
    public CursorRedisWriter cursorRedisWriter(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['params']}") String params) {
        CursorRedisWriter writer = new CursorRedisWriter();
        if (null == itemId) {
            return null;
        }
        WorkItem workItemDtl = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));
        writer.setItemDtl(workItemDtl);
        return writer;
    }


    /**
     * 自定义ProcessorLister
     *
     * @param itemId
     * @param instDtlId
     * @return
     */
    @Bean
    @StepScope
    public CursorProcessorLister cursorProcessorLister(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['instDtlId']}") Long instDtlId) {
        CursorProcessorLister processorLister = new CursorProcessorLister();
        if (null == itemId) {
            return null;
        }
        WorkItem workItemDtl = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));
        processorLister.setItemDtl(workItemDtl);
        if (null == instDtlId) {
            return null;
        }
        ScheduleInstanceDtl scheduleInstanceDtl = scheduleInstanceDtlDao.selectByPrimaryKey(instDtlId);
        processorLister.setInstanceDtl(scheduleInstanceDtl);

        return processorLister;
    }

    /**
     * 自定义WriterLister
     *
     * @param itemId
     * @param instDtlId
     * @return
     */
    @Bean
    @StepScope
    public CursorWriterLister cursorWriterLister(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['instDtlId']}") Long instDtlId) {
        CursorWriterLister cursorWriterLister = new CursorWriterLister();
        if (null == itemId) {
            return null;
        }
        WorkItem workItemDtl = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));
        cursorWriterLister.setItemDtl(workItemDtl);
        if (null == instDtlId) {
            return null;
        }
        ScheduleInstanceDtl scheduleInstanceDtl = scheduleInstanceDtlDao.selectByPrimaryKey(instDtlId);
        cursorWriterLister.setInstanceDtl(scheduleInstanceDtl);
        return cursorWriterLister;
    }

    /**
     * 自定义JobListener 生命周期batch job
     *
     * @param itemId
     * @param instDtlId
     * @return
     */
    @Bean
    @JobScope
    public CursorJobListener cursorJobListener(@Value("#{jobParameters['itemId']}") Long itemId, @Value("#{jobParameters['instDtlId']}") Long instDtlId) {
        CursorJobListener jobListener = new CursorJobListener();
        if (null == itemId) {
            return null;
        }
        WorkItem workItemDtl = workItemDao.selectByPrimaryKey(TypeCast.getLong(itemId));
        jobListener.setItemDtl(workItemDtl);
        if (null == instDtlId) {
            return null;
        }
        ScheduleInstanceDtl scheduleInstanceDtl = scheduleInstanceDtlDao.selectByPrimaryKey(instDtlId);
        jobListener.setInstanceDtl(scheduleInstanceDtl);

        Work work = workDao.selectByPrimaryKey(workItemDtl.getWorkId());
        jobListener.setWorkItem(work);
        return jobListener;
    }

    @Bean(name = "sqlJob")
    public Job itemDtlJob(JobBuilderFactory jobs, Step itemDtlStep, CursorJobListener cursorJobListener) throws Exception {
        return jobs.get("itemSqlJob").incrementer(new RunIdIncrementer()).listener(cursorJobListener).flow(itemDtlStep).end().build();
    }

    @Bean(name = "redisJob")
    public Job redisJob(JobBuilderFactory jobs, Step itemDtlRedisStep, CursorJobListener cursorJobListener) throws Exception {
        return jobs.get("itemRedisJob").incrementer(new RunIdIncrementer()).listener(cursorJobListener).flow(itemDtlRedisStep).end().build();
    }

    @Bean
    public Step itemDtlStep(StepBuilderFactory stepBuilderFactory, ItemReader reader, ItemWriter cursorWriter, ItemProcessor processor, CursorProcessorLister cursorProcessorLister, CursorWriterLister cursorWriterLister) throws Exception {
        return stepBuilderFactory.get("itemStep")
                .chunk(10000)
                .reader(reader)
                .processor(processor).listener(cursorProcessorLister)
                .writer(cursorWriter).listener(cursorWriterLister)
                .build();
    }

    @Bean
    public Step itemDtlRedisStep(StepBuilderFactory stepBuilderFactory, ItemReader reader, ItemWriter cursorRedisWriter, ItemProcessor processor, CursorProcessorLister cursorProcessorLister, CursorWriterLister cursorWriterLister) throws Exception {
        return stepBuilderFactory.get("itemStep")
                .chunk(10000)
                .reader(reader)
                .processor(processor).listener(cursorProcessorLister)
                .writer(cursorRedisWriter).listener(cursorWriterLister)
                .build();
    }


}
